package com.zyx.pulsardemo.protobuf;

import com.zyx.pulsardemo.protobuf.pojo.DogPOJO;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;

import java.util.Date;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author Yaxi.Zhang
 * @since 2021/10/21 14:36
 * desc: 发送 Protobuf 信息
 */
@SuppressWarnings("AlibabaAvoidManuallyCreateThread")
public class SendProtobufDemo {

    public static void main(String[] args) {
        SendProtobufDemo demo = new SendProtobufDemo();
        demo.startProducer();
    }

    private final String serviceUrl;
    protected String topic;

    private PulsarClient client;
    private Producer<DogPOJO.Dog> producer;

    public SendProtobufDemo() {
        this.serviceUrl = "pulsar://bigdata022:6650,bigdata023:6650,bigdata024:6650";
        this.topic = "persistent://manning/protobuf/example-topic";
    }

    protected void startProducer() {
        Runnable run = () -> {
            AtomicInteger ai = new AtomicInteger(1000);
            while (ai.get() < 1100) {
                try {
                    String date = new Date().toString();
                    DogPOJO.Dog dog = DogPOJO.Dog.newBuilder().setName("" + ai.getAndIncrement()).setDate(date).build();
                    System.out.println("send message =====>>>>> " + dog);
                    getProducer()
                            .newMessage()
                            .value(dog)
                            .send();
                    Thread.sleep(10);
                } catch (final Exception ex) {
                    ex.printStackTrace();
                }
            }
        };
        new Thread(run).start();
    }

    protected PulsarClient getClient() throws PulsarClientException {
        if (client == null) {
            synchronized(this) {
                client = PulsarClient.builder().serviceUrl(serviceUrl).build();
            }
        }
        return client;
    }

    protected Producer<DogPOJO.Dog> getProducer() throws PulsarClientException {
        if (producer == null) {
            synchronized(this) {
                producer = getClient().newProducer(Schema.PROTOBUF(DogPOJO.Dog.class)).topic(topic).create();
            }
        }
        return producer;
    }

}
